import java.text.SimpleDateFormat

import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import java.util.{Date, Locale}

object LogAnalysis {

  val INPUT_TIME_FORMAT = FastDateFormat.getInstance("dd/MMM/yyyy:HH:mm:ss Z", Locale.ENGLISH)

  //目标日期格式
  val TARGET_FOMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")

  def parseTime(time: String) = {
    TARGET_FOMAT.format(new Date(getTime(time)))
  }

  def getTime(time: String) = {
    try {
      INPUT_TIME_FORMAT.parse(time.substring(time.indexOf("[") + 1, time.lastIndexOf("]"))).getTime()
    } catch {
      case e: Exception => {
        0L
      }
    }
  }

  def getHour(timelong: Long): String = {
    import org.joda.time.DateTime
    val datetime = new DateTime(timelong.toLong)
    datetime.getHourOfDay.toString
  }

  /**
   * 时间字符串转毫秒时间戳
   *
   * @param tm eg：“2020-09-04 00:00:00”
   * @return 1599148800000
   */
  def tranTimeToLong(tm: String): Long = {
    val fm = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val dt: Date = fm.parse(tm)
    val tim: Long = dt.getTime()
    tim
  }

  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")


    /**
     * IP
     * 命中率(Hit/Miss)
     * 响应时间
     * 请求时间
     * 请求方法
     * 请求URL
     * 请求协议
     * 状态码
     * 响应大小
     * referer
     * 用户代理
     */


    val str = s"""111.19.97.15 HIT 18 [15/Feb/2017:00:00:39 +0800] "GET http://cdn.v.abc.com.cn/videojs/video-js.css HTTP/1.1" 200 14727 "http://www.zzqbsm.com/" "Mozilla/5.0+(Linux;+Android+5.1;+vivo+X6Plus+D+Build/LMY47I)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Version/4.0+Chrome/35.0.1916.138+Mobile+Safari/537.36+T7/7.4+baiduboxapp/8.2.5+(Baidu;+P1+5.1)"""
    //    val fields = str.split("\\s+")
    //    fields.foreach(println)

    /**
     * 111.19.97.15
     * HIT
     * 18
     * [15/Feb/2017:00:00:39
     * +0800]
     * "GET
     * http://cdn.v.abc.com.cn/videojs/video-js.css
     * HTTP/1.1"
     * 200
     * 14727
     * "http://www.zzqbsm.com/"
     * "Mozilla/5.0+(Linux;+Android+5.1;+vivo+X6Plus+D+Build/LMY47I)+AppleWebKit/537.36+(KHTML,+like+Gecko)+Version/4.0+Chrome/35.0.1916.138+Mobile+Safari/537.36+T7/7.4+baiduboxapp/8.2.5+(Baidu;+P1+5.1)
     *
     */

    //从文件中读取数据
        val loglines: RDD[String] = sc.textFile("file:///F:\\lagou\\lagouhomework\\stage_4_module_2\\2.日志分析\\LogAnalysis\\data\\cdn.txt")
//    val loglines: RDD[String] = sc.textFile("file:////Users/yaong/Documents/Develop/lagouhomework/stage_4_module_2/2.日志分析/LogAnalysis/data/cdn.txt")
    println(s"源日志总行数 = ${loglines.count}")
    /**
     * 源日志总行数 = 265831
     */

    /////////////////////////////1.计算独立IP数/////////////////////////////
    val logData = loglines.map { line =>
      val field = line.split("\\s+")
      (field(0), 1)
    }.reduceByKey(_ + _).sortBy(_._2, false)
    //1.1 统计IP个数
    println(s"统计IP总数 = ${logData.count}")
    /**
     * 统计IP总数 = 43649
     */

    //1.1 展示前10个频次最高的IP和次数
    val top10 = logData.take(10).foreach(println)
    /**
     * (114.55.227.102,9348)
     * (220.191.255.197,2640)
     * (115.236.173.94,2476)
     * (183.129.221.102,2187)
     * (112.53.73.66,1794)
     * (115.236.173.95,1650)
     * (220.191.254.129,1278)
     * (218.88.25.200,751)
     * (183.129.221.104,569)
     * (115.236.173.93,529)
     */

    /////////////////////////////2.统计每个视频独立IP数/////////////////////////////
    //规律：*.mp4在访问方式后面一个字段，且字段能独立解析到，
    // "GET http://cdn.v.abc.com.cn/140987.mp4 HTTP/1.1"

    // 分析所有log，得出 ip和请求URL
    val videoRDD = loglines.map { line =>
      val field = line.split("\\s+")
      (field(0), field(6))
    }
    // 过滤所有log，得出请求URL字段包含“.mp4”的条目
    val mp4RDD = videoRDD.filter(_._2.takeRight(4) == ".mp4")
    println(s"包含MP4的url数 = ${mp4RDD.count}")
    /**
     * 包含MP4的url数 = 137508
     */

    //去重
    val distinctRDD = mp4RDD.distinct
    println(s"去重后的url数 = ${distinctRDD.count}")
    /**
     * 去重后的url数 = 38639
     */

    //变形
    val exchangeRDD = distinctRDD.map { case (ip, url) => (url, ip) }
    exchangeRDD.take(3) foreach (println)
    /**
     * (http://v-cdn.abc.com.cn/89973.mp4,1.204.28.15)
     * (https://v-cdn.abc.com.cn/140822.mp4,101.86.58.73)
     * (https://v-cdn.abc.com.cn/89973.mp4,113.67.144.183)
     */

    //统计每个url对应的ip数
    val countRDD = distinctRDD.map { case (ip, url) => (url, 1) }.reduceByKey(_ + _).sortBy(_._2, false)
    //打印前三个访问较多的
    countRDD.take(3).foreach(println)

    /**
     * (http://v-cdn.abc.com.cn/141081.mp4,2051)
     * (http://v-cdn.abc.com.cn/141032.mp4,1107)
     * (http://v-cdn.abc.com.cn/140995.mp4,1088)
     */

    ///////////////////////////////////////////////////////////////////////////////////////
    /////////////////////////////3.统计一天中每个小时的流量/////////////////////////////
    ///////////////////////////////////////////////////////////////////////////////////////

    // 解析时间字符串
    // https://blog.csdn.net/cuiyang0720/article/details/98599470
    println(parseTime("[10/Nov/2016:00:01:02 +0800]"))
    /**
     * 2016-11-10 00:01:02
     */

    // 分析所有log，得出 ip和请求URL
    val timeRDD = loglines.map { line =>
      val field = line.split("\\s+")
      (field(0), field(6), field(3) + " " + field(4))
    }
    // 过滤所有log，得出请求URL字段包含“.mp4”的条目
    val mp4RDD2 = timeRDD.filter(_._2.takeRight(4) == ".mp4")
    println(s"包含MP4的url数 = ${mp4RDD2.count}")
    mp4RDD2.collect().take(3).foreach(println)
    /**
     * 包含MP4的url数 = 137508
     */

    val hourRDD = mp4RDD2.map { case (ip, url, timeStr) =>
      val datetime = parseTime(timeStr);
      //      println(datetime)
      val longtime = tranTimeToLong(datetime)
      //      println(longtime)
      val hour = getHour(longtime)
      //      println(hour)
      (url, ip, hour)
    }

    /**
     * 2017-02-15 21:42:30
     * 1487166150000
     * 21
     */
    hourRDD.collect().take(3).foreach(println)
    //去重,排除一小时内，同一IP多次访问一个视频的重复log
    val distinctHour = hourRDD.distinct
    println(s"去重后的url数 = ${distinctHour.count}")
    /**
     * 去重后的url数 = 42035
     */

    //转换为 小时和统计次数的关系，次数=同一小时内，若干url次数之和
    val sumRDD = distinctHour.map { case (url, ip, hour) => ((url, hour), 1) }.reduceByKey(_ + _) //.map{case ((url, hour), count) => ((url, hour), (ip, count))}
    sumRDD.collect().take(3).foreach(println)
    /**
     * ((http://v-cdn.abc.com.cn/138380.mp4,8),3)
     * ((https://v-cdn.abc.com.cn/84907.mp4,9),1)
     * ((http://v-cdn.abc.com.cn/140822.mp4,16),9)
     */

    //hour转成整数，便于排序，同时累加同一个小时内，所有url的次数求和
    val hourSum = sumRDD.map { case ((url, hour), cnt) => (hour.toInt, cnt) }.reduceByKey(_ + _).sortBy(_._1)
    hourSum.collect().foreach(println)

    /**
     * (0,505)
     * (1,274)
     * (2,196)
     * (3,169)
     * (4,169)
     * (5,335)
     * (6,597)
     * (7,1241)
     * (8,2385)
     * (9,2487)
     * (10,2505)
     * (11,2378)
     * (12,2493)
     * (13,2898)
     * (14,2961)
     * (15,2773)
     * (16,2840)
     * (17,2350)
     * (18,2121)
     * (19,2322)
     * (20,2494)
     * (21,2439)
     * (22,1981)
     * (23,1122)
     */
    // 关闭SparkContext
    sc.stop()
  }
}
